package com.atguigu.userprofile.app;

import com.atguigu.bean.TagInfo;
import com.atguigu.bean.TaskInfo;
import com.atguigu.bean.TaskTagRule;
import com.atguigu.constans.ConstCodes;
import com.atguigu.dao.TagInfoDAO;
import com.atguigu.dao.TaskInfoDAO;
import com.atguigu.dao.TaskTagRuleDAO;
import com.atguigu.util.MyPropertiesUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

public class TaskSQLAPP {
    public static void main(String[] args) {
        /**
         * 1.获取到taskId和业务日期
         * 2.获取tag_info、task_Info、task_tag_rule这个三张表的数据
         * 3.动态拼写建表语句
         * 4.动态拼写查询语句（基于之前标签任务中定义好的sql，在这个基础上去拼写最终需要的查询语句）
         * 5.拼写插入语句
         * 6.执行
         */

        //注意！！！ 如果打包到集群运行的话要注释掉setMaster
        SparkConf sparkConf = new SparkConf().setAppName("TaskSQLAPP");//.setMaster("local[*]");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        //0.获取相关参数
        Properties properties = MyPropertiesUtil.load("config.properties");
        String hdfsPath = properties.getProperty("hdfs-store.path");
        String dwDBName = properties.getProperty("data-warehouse.dbname");
        String upDBName = properties.getProperty("user-profile.dbname");

        //1.获取到taskId和业务日期
        String taskId = args[0];
        String busiDate = args[1];

        //2.获取tag_info、task_Info、task_tag_rule这个三张表的数据
        TaskInfo taskInfo = TaskInfoDAO.getTaskInfo(taskId);
        List<TaskTagRule> taskTagRuleList = TaskTagRuleDAO.getTaskTagRule(taskId);
        TagInfo tagInfo = TagInfoDAO.getTagInfo(taskId);

        //3.动态拼写建表语句
        /**
         * create table if not exists 表名
         *   (uid String,tag_value 类型)
         *  partitioned by (dt String)
         *   comment ‘xxx'
         *   ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
         *   location ‘路径’
         */

        //3.1获取表名
        String tableName = tagInfo.getTagCode().toLowerCase();
//        System.out.println(tableName);


        String fileType = null;
        if (ConstCodes.TAG_VALUE_TYPE_LONG.equals(tagInfo.getTagValueType())){
            fileType = "bigint";
        }else if (ConstCodes.TAG_VALUE_TYPE_DECIMAL.equals(tagInfo.getTagValueType())){
            fileType = "decimal";
        }else if (ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagInfo.getTagValueType())){
            fileType = "String";
        }else if (ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagInfo.getTagValueType())){
            fileType = "String";
        }

        String createTableSQL = "    create table if not exists "+upDBName+"."+tableName+"\n" +
                "            (uid String,tag_value "+fileType+")\n" +
                "            comment '"+tagInfo.getTagName()+"' \n" +
                "           partitioned by (dt String)\n" +
                "            ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t' \n" +
                "           location '"+hdfsPath+"/"+upDBName+"/"+tableName+"'";

//        System.out.println(createTableSQL);

        // 4.动态拼写查询语句（基于之前标签任务中定义好的sql，在这个基础上去拼写最终需要的查询语句）
        /**
         * select
         *  uid,
         *  case query_value when 'F' then '男'  when 'M' then '女'  when 'U' then '未知'  end as query_value
         *  from
         *  (
         *    select
         *     id as uid,
         *     if(gender<>"",gender,"U")  as query_value
         *    from
         *     dim_user_zip where dt='9999-12-31'
         *   )ti
         */

        //获取子查询
        String subQuerySQL = taskInfo.getTaskSql();

        //动态拼接WhenThen语句
        List<String> whenThenSQLList = taskTagRuleList.stream().map(taskTagRule -> "when '" + taskTagRule.getQueryValue() + "' then '" + taskTagRule.getSubTagValue() + "'").collect(Collectors.toList());

//        System.out.println(whenThenSQLList);
        String whenThenSQL = StringUtils.join(whenThenSQLList, " ");
//        System.out.println(whenThenSQL);


        String querySQL = null;
        //如果有四级标签那么需要做case when
        if (taskTagRuleList.size()>0){
            querySQL = "  select \n" +
                    "           uid, \n" +
                    "           case query_value "+whenThenSQL+"  end as tag_value\n" +
                    "          from\n" +
                    "           ("+subQuerySQL+")ti";
        }else {
            //如果没有四级标签则不需要case when
            querySQL = "  select \n" +
                    "           uid, \n" +
                    "          query_value  as tag_value\n" +
                    "          from\n" +
                    "           ("+subQuerySQL+")ti";
        }

        //5.拼写插入语句
        String insertSQL = "insert overwrite table " +upDBName+"."+ tableName + " partition(dt='" + busiDate + "') " + querySQL;
        System.out.println(insertSQL);

        //6.执行sql
        //默认都在gmall库中执行
        sparkSession.sql("use " + dwDBName);
        sparkSession.sql(createTableSQL);
        sparkSession.sql(insertSQL);

    }
}
